package com.bkd.report

import com.bkd.beans.Log
import com.bkd.tools.ReqUtil
import com.bkd.util.JedisPool
import org.apache.commons.lang.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedis

object AppRpt2 extends App{

  if(args.length!=2) {
    println(
      """
        |com.bkd.report.AppRpt2
        |参数
        |输入文件路径
        |输出文件路径
      """.stripMargin)
    sys.exit()
  }

  //制定接受的参数
  val Array(inputPath,outputPath)=args
  //创建sparkconf
  private val conf = new SparkConf()
  conf.setAppName(s"${this.getClass.getSimpleName}")
  conf.setMaster("local[*]")
  //制定序列化方式
  conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  //创建sparkContext
  private val sc = new SparkContext(conf)

  sc.textFile(inputPath)
    .map(_.split(",",-1))
    .filter(_.length>=85)
    .map(Log(_))
    .filter(log=> !log.appid.isEmpty || !log.appname.isEmpty)
  // 从redis中，查询所有的key对应的value,并进行转换。
    .mapPartitions(itr => {
    //  封装一个迭代器。
    var parList = new collection.mutable.ListBuffer[(String,List[Double])]()
    val jedis: Jedis = JedisPool.getJedis()
    //  读取log所有的appid.到redis中，进行key-value的比较。
    itr.foreach(log=>{
      var appName = log.appname
      if(!StringUtils.isNotEmpty(appName)){
        appName = jedis.get(log.appid)
      }

      //计算其他指标
      val reqList: List[Double] = ReqUtil.caculateReq(log.requestmode,log.processnode)
      val parceList: List[Double] = ReqUtil.caculatePrc(log.iseffective,log.isbilling,log.isbid,log.adorderid,log.iswin,log.winprice,log.adpayment)
      val clickList: List[Double] = ReqUtil.caculateClick(log.requestmode,log.iseffective)
      // 组合元组。
      parList+=((appName,reqList++parceList++clickList))
    })
    parList.iterator
  }).reduceByKey((list1,list2)=>{
    list1.zip(list2).map(t=>t._1+t._2)
  }).map(t=>t._1+","+t._2.mkString(",")).saveAsTextFile(outputPath)
  sc.stop()
}
